package com.xkcoding.mq.kafka.handler;

import com.alibaba.fastjson.JSONObject;
import com.xkcoding.mq.kafka.constants.KafkaConsts;
import com.xkcoding.mq.kafka.mapper.UserMapper;
import com.xkcoding.mq.kafka.module.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.ExecutorType;
import org.apache.ibatis.session.SqlSession;
import org.mybatis.spring.SqlSessionTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.CountDownLatch;

@Component
@Slf4j
public class SyncDataHandler {
    @Autowired
    ThreadPoolTaskExecutor asyncThreadPool;
    @Autowired
    SqlSessionTemplate sqlSessionTemplate;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    private UserMapper userMapper;


    public void testSend() throws InterruptedException {
        int batchSize = 100;
        CountDownLatch countDownLatch = new CountDownLatch(batchSize);

        try {
            User user = new User();
            user.setAge(1);
            user.setName("hg");
            user.setSex(1);
            for (int i = 0; i < batchSize; i++) {
                asyncThreadPool.execute(()->{
                    testInsert(100,user);
                    countDownLatch.countDown();
                });
            }

        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            countDownLatch.await();
            kafkaTemplate.send(KafkaConsts.TOPIC_TEST, "同步完毕...");
        }
    }


    private void testInsert(int size, User user){
        SqlSession sqlSession = sqlSessionTemplate.getSqlSessionFactory().openSession(ExecutorType.BATCH, false);
        UserMapper mapper = sqlSession.getMapper(UserMapper.class);
        try {
            for (int i = 0; i < size; i++) {
                mapper.insert(user);
            }
            sqlSession.commit();
            sqlSession.clearCache();
        }catch (Exception e){
            sqlSession.rollback();
        }
        finally {
            sqlSession.close();
        }
    }


    public void sendData(){
        int count = userMapper.selectUserTableCount();
        int pageSize = 100000;
        int pageNum = count/pageSize;
        for (int i = 1; i <= 1; i++) {
            List<User> users = userMapper.selectUsers((i - 1) * pageSize, pageSize);
            kafkaTemplate.send(KafkaConsts.TOPIC_TEST, JSONObject.toJSONString(users));
        }
    }



}
